package spring.cloud.tasks.executor.executor;

import com.google.common.collect.Maps;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import spring.cloud.tasks.executor.exception.JobException;
import spring.cloud.tasks.executor.exception.JobInitAlarmException;
import spring.cloud.tasks.executor.listener.config.ConfigurationNode;
import spring.cloud.tasks.executor.listener.config.TaskConfiguration;
import spring.cloud.tasks.executor.node.TaskNodePaths;
import spring.cloud.tasks.executor.task.TaskExecutor;
import spring.cloud.tasks.executor.task_type.TaskType;
import spring.cloud.tasks.executor.task_type.TaskTypeManager;
import spring.cloud.tasks.executor.utils.AlarmUtils;
import spring.cloud.tasks.executor.utils.SystemEnvProperties;
import spring.cloud.tasks.executor.utils.log.LogUtils;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;

@Slf4j
@Setter
@Getter
class TaskInitListener implements TreeCacheListener {

    private final TaskInitService taskInitService;

    public TaskInitListener(TaskInitService taskInitService) {
        this.taskInitService = taskInitService;
    }

    @Override
    public void childEvent(CuratorFramework curatorFramework, TreeCacheEvent treeCacheEvent) throws Exception {
        if (treeCacheEvent == null) {
            return;
        }
        ChildData childData = treeCacheEvent.getData();
        if (childData == null) {
            return;
        }
        String path = childData.getPath();
        if (path == null || path.equals(TaskNodePaths.ROOT)) {
            return;
        }
        TreeCacheEvent.Type type = treeCacheEvent.getType();
        if (type == null || !type.equals(TreeCacheEvent.Type.NODE_ADDED)) {
            return;
        }
        String taskId = substringAfterLast(path, "/");
        String taskClassPath = TaskNodePaths.getTaskNodePath(taskId, ConfigurationNode.TASK_CLASS);
        // wait 5 seconds at most until jobClass created.
        for (int i = 0; i < ExecutorAndTaskInitService.WAIT_JOBCLASS_ADDED_COUNT; i++) {
            if (!taskInitService.zookeeper.checkExists(taskClassPath)) {
                Thread.sleep(200L);
                continue;
            }
            LogUtils.info(log, taskId, "new job: {} 's jobClass created event received", taskId);
            if (!taskInitService.taskIdSet.contains(taskId)) {
                if (initTaskExecutor(taskId)) {
                    taskInitService.taskIdSet.add(taskId);
                    LogUtils.info(log, taskId, "the job {} initialize successfully", taskId);
                }
            } else {
                LogUtils.warn(log, taskId, "the job {} is unnecessary to initialize, because it's already existing", taskId);
            }
            break;
        }
    }

    private String substringAfterLast(String path, String s) {
        return null;
    }


    private boolean initTaskExecutor(String taskId) {
        try {
            LogUtils.info(log, taskId, "start to initialize the new job");
            TaskInitService.JOB_INIT_FAILED_RECORDS.get(taskInitService.executorId).putIfAbsent(taskId, new HashSet<Integer>());
            TaskConfiguration taskConfiguration = new TaskConfiguration(taskId, taskInitService.zookeeper);
            TaskType taskType = TaskTypeManager.get(taskConfiguration.getTaskType());
            if (taskType == null) {
                String message = String.format("the jobType %s is not supported by the executor version %s", taskConfiguration.getTaskType(), "");
                LogUtils.warn(log, taskId, message);
                throw new JobInitAlarmException(message);
            }
            if (taskType.getHandlerClass() == null) {
                throw new JobException("unexpected error, the saturnJobClass cannot be null, jobName is %s, jobType is %s", taskId, taskConfiguration.getTaskType());
            }
            if (taskConfiguration.isDeleting()) {
                String executorNodePath = TaskNodePaths.getExecutorNodePath(taskId, taskInitService.executorId);
                taskInitService.zookeeper.delete(executorNodePath);
                LogUtils.warn(log, taskId, "the job is on deleting");
                return false;
            }
            TaskExecutor taskExecutor = new TaskExecutor(taskInitService.zookeeper, taskConfiguration);
            taskExecutor.setTaskInitService(taskInitService.taskInitService);
            taskExecutor.init();
            // clear previous records when initialize job successfully
            TaskInitService.JOB_INIT_FAILED_RECORDS.get(taskInitService.executorId).get(taskId).clear();
            return true;
        } catch (JobInitAlarmException e) {
            if (!SystemEnvProperties.VIP_SATURN_DISABLE_JOB_INIT_FAILED_ALARM) {
                // no need to log exception stack as it should be logged in the original happen place
                raiseAlarmForJobInitFailed(taskId, e);
            }
        } catch (Throwable t) {
            LogUtils.warn(log, taskId, "job initialize failed, but will not stop the init process", t);
        }

        return false;
    }

    private void raiseAlarmForJobInitFailed(String jobName, JobInitAlarmException jobInitAlarmException) {
        String message = jobInitAlarmException.getMessage();
        int messageHashCode = message.hashCode();
        Set<Integer> records = TaskInitService.JOB_INIT_FAILED_RECORDS.get(taskInitService.executorId).get(jobName);
        if (!records.contains(messageHashCode)) {
            try {
                String namespace = taskInitService.zookeeper.getNameSpace();
                AlarmUtils.raiseAlarm(constructAlarmInfo(namespace, jobName, taskInitService.executorId, message), namespace);
                records.add(messageHashCode);
            } catch (Exception e) {
                LogUtils.error(log, jobName, "exception throws during raise alarm for job init fail", e);
            }
        } else {
            LogUtils.info(log, jobName,
                    "job initialize failed but will not raise alarm as such kind of alarm already been raise before");
        }
    }

    private Map<String, Object> constructAlarmInfo(String namespace, String jobName, String executorName, String alarmMessage) {
        Map<String, Object> alarmInfo = new HashMap<>();

        alarmInfo.put("jobName", jobName);
        alarmInfo.put("executorName", executorName);
        alarmInfo.put("name", "Saturn Event");
        alarmInfo.put("title", String.format("JOB_INIT_FAIL:%s", jobName));
        alarmInfo.put("level", "CRITICAL");
        alarmInfo.put("message", alarmMessage);

        Map<String, String> customFields = Maps.newHashMap();
        customFields.put("sourceType", "saturn");
        customFields.put("domain", namespace);
        alarmInfo.put("additionalInfo", customFields);

        return alarmInfo;
    }

}
